package com.atguigu.edu.app.dws;

import bean.TradeSourceOrderBean;
import bean.TradeTotalBean;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import util.MyClickhouseUtil;
import util.MyKafkaUtil;

import java.util.Collections;
import java.util.HashSet;

/**
 * 交易综合统计
 */
public class DwsTradeTotalOrderWindow {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);

        String topic = "dwd_trade_order_detail";
        String group = "dws_trade_total_order_window";
        DataStreamSource<String> sourceDS = env.addSource(MyKafkaUtil.getKafkaConsumer(topic, group));

        // 类型转换
        SingleOutputStreamOperator<TradeTotalBean> mapDS = sourceDS.map(jsonStr -> {
            JSONObject jsonObject = JSON.parseObject(jsonStr);
            String orderId = jsonObject.getString("order_id");
            String userId = jsonObject.getString("user_id");
            Double orderAmount = jsonObject.getDouble("final_amount");
            long ts = jsonObject.getLong("od_ts") * 1000L;
            return TradeTotalBean.builder().
                    orderAmount(orderAmount)
                    .userIdSet(new HashSet<>(Collections.singleton(userId)))
                    .orderIdSet(new HashSet<>(Collections.singleton(orderId)))
                    .ts(ts)
                    .build();
        });
        SingleOutputStreamOperator<TradeTotalBean> withWatermarkDS = mapDS.assignTimestampsAndWatermarks(
                WatermarkStrategy
                        .<TradeTotalBean>forMonotonousTimestamps()
                        .withTimestampAssigner((data, ts) -> data.getTs())
        );

        KeyedStream<TradeTotalBean, Tuple> keyedDS = withWatermarkDS.keyBy("ll");

        WindowedStream<TradeTotalBean, Tuple, TimeWindow> windowDS = keyedDS.window(TumblingEventTimeWindows.of(Time.seconds(10L)));

        SingleOutputStreamOperator<TradeTotalBean> reduceDS = windowDS.reduce(
                new ReduceFunction<TradeTotalBean>() {
                    @Override
                    public TradeTotalBean reduce(TradeTotalBean value1, TradeTotalBean value2) throws Exception {
                        value1.getOrderIdSet().addAll(value2.getOrderIdSet());
                        value1.getUserIdSet().addAll(value2.getUserIdSet());
                        value1.setOrderAmount(value1.getOrderAmount() + value2.getOrderAmount());
                        return value1;
                    }
                },
                new ProcessWindowFunction<TradeTotalBean, TradeTotalBean, Tuple, TimeWindow>() {
                    @Override
                    public void process(Tuple tuple, Context context, Iterable<TradeTotalBean> elements, Collector<TradeTotalBean> out) throws Exception {
                        long stt = context.window().getStart();
                        long edt = context.window().getEnd();
                        for (TradeTotalBean element : elements) {
                            element.setUserCount((long) element.getUserIdSet().size());
                            element.setOrderCount((long) element.getOrderIdSet().size());
                            element.setStt(stt);
                            element.setEdt(edt);
                            element.setTs(System.currentTimeMillis());
                            out.collect(element);
                        }
                    }
                }
        );
        SinkFunction<TradeTotalBean> sinkFunction = MyClickhouseUtil.<TradeTotalBean>getSinkFunction(
                "\"insert into dws_trade_total_order_window values(?,?,?,?,?,?)\""
        );
        reduceDS.addSink(sinkFunction);
        env.execute();

    }
}
